﻿#ifdef LINUX

#include "Network/TCPClient.h"
#include "Async/PackagedTask.h"

#include "Platform/Socket/SocketPlatform.h"

#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>

//memset
#include <cstring>

void CTCPClient::Connect(const SIPAddress &IP, uint16_t Port)
{
    if(State!=ETCPClientState::Idle)
    {
        OnErrorOccured.ExecuteIfBound(ETCPClientError::ConnectTwiceError);
        return;
    }


    //创建socket,use Linux api
    PlatformData->Socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (PlatformData->Socket < 0)
    {
        OnErrorOccured.ExecuteIfBound(ETCPClientError::CreateSocketError);
        return;
    }

    sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(Port);
    if (IP.IsIPv6())
    {
        memcpy(&addr.sin_addr, IP.IP.IPv6, sizeof(IP.IP.IPv6));
    }
    else
    {
        memcpy(&addr.sin_addr, IP.IP.IPv4, sizeof(IP.IP.IPv4));
    }

    int32_t Result = connect(PlatformData->Socket, (sockaddr*)&addr, sizeof(addr));
    if(Result != 0)
    {
        OnErrorOccured.ExecuteIfBound(ETCPClientError::ConnectError);
        close(PlatformData->Socket);
        PlatformData->Socket = -1;
        return;
    }

    //将socket设置为非阻塞
    int Flags = fcntl(PlatformData->Socket, F_GETFL, 0);
    fcntl(PlatformData->Socket, F_SETFL, Flags | O_NONBLOCK);

    State=ETCPClientState::Connected;

    PostConnected();
}

void CTCPClient::Disconnect()
{
    std::lock_guard<std::mutex> Lock(SocketMutex);
    if(PlatformData->Socket!=-1)
    {
        close(PlatformData->Socket);
        PlatformData->Socket=-1;
        State=ETCPClientState::Idle;
        OnDisconnected.ExecuteIfBound();
    }
}


void CTCPClient::StartReceiveAndDetectTask()
{
    //创建一个异步任务，用于接收数据
    TThreadSafeWeakPtr<CTCPClient> WeakThis= AsShared();

    std::shared_ptr<CPackagedTask<void>> ReciveTask
    = std::make_shared<CPackagedTask<void>>(
        CAsyncSystem::Get()->ThreadPool
        ,[WeakThis](bool&)  mutable ->void
    {
        auto This=WeakThis.Lock();
        if(!This)
        {
            //客户端已被销毁
            return;
        }

        TVector<uint8_t> Data;
        int32_t RevicedByteCount;
        {
            std::lock_guard<std::mutex> Lock(This->SocketMutex);
            if(This->PlatformData->Socket==-1)
            {
                //socket已经关闭
                return;
            }

            //读取数据
            Data.Resize(This->ReceiveBufferSize);
            RevicedByteCount = recv(This->PlatformData->Socket, Data.Data(), Data.Num(), 0);
        
        }


        if(RevicedByteCount==0)
        {
            //连接被正常关闭
            std::lock_guard<std::mutex> Lock(This->SocketMutex);
            if(This->State!=ETCPClientState::Idle)
            {
                close(This->PlatformData->Socket);
                This->PlatformData->Socket=-1;
                This->State=ETCPClientState::Idle;
                This->OnDisconnected.ExecuteIfBound();
            }
            return;
        }

        if(RevicedByteCount<0)
        {
            if(errno==EWOULDBLOCK)
            {
                //没有数据,继续等待
                This->StartReceiveAndDetectTask();
                return;
            }

            if(errno==EMSGSIZE)
            {
                //缓冲区太小
                This->ReceiveBufferSize=This->ReceiveBufferSize.load()*2;
                This->StartReceiveAndDetectTask();
                return;
            }

          
            //这个socket出现了错误
            std::lock_guard<std::mutex> Lock(This->SocketMutex);
            close(This->PlatformData->Socket);
            This->PlatformData->Socket=-1;
            This->State=ETCPClientState::Idle;
            This->OnErrorOccured.ExecuteIfBound(ETCPClientError::ConnectInterruptedError);
            return;
        }

        //一切正常，继续接收
        This->StartReceiveAndDetectTask();

        //处理接收到的数据
        Data.Resize(RevicedByteCount);
        {
            std::lock_guard<std::mutex> Lock(This->ReceivedDataMutex);
            This->ReceivedData.Append(Data);
        }
        This->OnDataReceived.ExecuteIfBound();
        
    });

    ReciveTask->Start();
}

void CTCPClient::StartSendTask()
{
    //创建一个异步任务，用于发送数据
    TThreadSafeWeakPtr<CTCPClient> WeakThis= AsShared();

    std::shared_ptr<CPackagedTask<void>> SendTask
    = std::make_shared<CPackagedTask<void>>(
        CAsyncSystem::Get()->ThreadPool
        ,[WeakThis](bool&)  mutable ->void
    {
        auto This=WeakThis.Lock();
        if(!This)
        {
            //客户端已被销毁
            return;
        }

        if(This->State!=ETCPClientState::Connected)
        {
            //连接已经断开
            return;
        }

        bool bHasData=false;
        {
            std::lock_guard<std::mutex> Lock(This->SendDataMutex);
            bHasData=!This->SendData.Empty();
        }

        if(!bHasData)
        {
            //没有数据需要发送,继续等待
            This->StartSendTask();
            return;
        }


        int32_t SendByteCount;
        {
            std::lock_guard<std::mutex> Lock(This->SocketMutex);
            if(This->PlatformData->Socket==-1)
            {
                //socket已经关闭
                return;
            }

            //发送数据
            SendByteCount= send(This->PlatformData->Socket, (char*)This->SendData.Data(), This->SendData.Num(), 0);
            //SendByteCount = send((SOCKET)This->Socket, (char*)This->SendData.Data(), This->SendData.Num(), 0);
        }

        if(SendByteCount==0)
        {
            //连接被正常关闭
            std::lock_guard<std::mutex> Lock(This->SocketMutex);
            if(This->State!=ETCPClientState::Idle)
            {
                close(This->PlatformData->Socket);
                This->State=ETCPClientState::Idle;
                This->OnDisconnected.ExecuteIfBound();
            }
        }

        if(SendByteCount<0)
        {
            if(errno==EWOULDBLOCK)
            {
                //数据发送缓冲区已满,继续等待
                This->StartSendTask();
                return;
            }

            //这个socket出现了错误
            std::lock_guard<std::mutex> Lock(This->SocketMutex);
            close(This->PlatformData->Socket);
            This->State=ETCPClientState::Idle;
            This->OnErrorOccured.ExecuteIfBound(ETCPClientError::ConnectInterruptedError);
            return;
        }

        //一切正常，删除已经发送的数据
        {
            std::lock_guard<std::mutex> Lock(This->SendDataMutex);
            This->SendData.Remove(0,SendByteCount,false);
        }

        //继续发送
        This->StartSendTask();
    });

    SendTask->Start();
}

void CTCPClient::InitWith(CSocketPlatformData*  InConnectedSocket)
{
    PlatformData=InConnectedSocket;
    State=ETCPClientState::Connected;

    //将socket设置为非阻塞
    int Flags = fcntl(PlatformData->Socket, F_GETFL, 0);
    fcntl(PlatformData->Socket, F_SETFL, Flags | O_NONBLOCK);


    PostConnected();
}

#endif